/*******************************************************************/
/* @ All Rights Reserved Copyright (C) 2014 broada,Co.,Ltd.        */
/*                                                                 */
/*  NAME      = activemq_core.c                                    */
/*  NOTE      =                                                    */
/*  DATE      = 2015/04/28 by geyh                                 */
/*******************************************************************/
#include <stdio.h>
#include <stdlib.h>

#include "frcomm.h"
#include "frreport_comm.h"
#include "activemq_openwire_parse.h"
#include "activemq_base_type.h"
#include "activemq_common.h"
#include "activemq_parse.h"
#include "frct_comm.h"

/*******************************************************************************
*  FUNC     :  activeMQ 各类数据块的解析
*  ARGS     :  p_st_data                   输入力数据
*           :  p_st_body_data                上报链表
*           :  p_st_private                私有数据
*           :  p_st_session                session数据
*  RTN      :  APM_L7_ANALYZE_FAIL         失败
*           :  APM_L7_ANALYZE_SUCC         成功
*           :  APM_L7_ANALYZE_TODO         数据解析未结束
*  NOTE     :
*******************************************************************************/
int activeMQ_object_parse(Frbyte_Packet *p_st_data, Frreport_body_data *p_st_body_data, L7session_activeMQ_private *p_st_private, L7_session *p_st_session)
{
    int i_ret = APM_L7_ANALYZE_SUCC;
    char arr_type;

    if(p_st_data == NULL || p_st_body_data == NULL || p_st_private == NULL || p_st_session == NULL)
        return APM_L7_ANALYZE_FAIL;

    i_ret = bytebuff_get_char(p_st_data, (uint8_t *)&arr_type);
    if(i_ret == FR_FAIL)
        return APM_L7_ANALYZE_FAIL;

    switch(arr_type)
    {
        /*
        * Info objects sent back and forth client/server when
        * setting up a client connection.
        */
        case ACTIVEMQ_WIREFORMAT_INFO:
            //将i_flag_try置0
            p_st_private->i_flag_try = 0;
            p_st_private->i_flag_responserequired = 1;
            i_ret = openwire_wireinfo_tightUnmarshal(p_st_data, p_st_body_data, p_st_private->i_data_len - 1, p_st_private);
            break;

//        case ACTIVEMQ_BROKER_INFO:
//            i_ret = openwire_brokerinfo_tightUnmarshal(p_st_data, p_st_body_data, p_st_private);
//            break;

        case ACTIVEMQ_CONNECTION_INFO:
            p_st_private->i_flag_data_type = FRLOGIN_START;
            i_ret = openwire_connectioninfo_tightUnmarshal(p_st_data, p_st_body_data, p_st_private);

            break;
//
//        case ACTIVEMQ_SESSION_INFO:
//            i_ret = openwire_sessioninfo_tightUnmarshal(p_st_data, p_st_body_data, p_st_private);
//            break;

        case ACTIVEMQ_CONSUMER_INFO:
            p_st_private->is_producer = 2;
            p_st_private->i_flag_report = 1;
            i_ret = openwire_consumerinfo_tightUnmarshal(p_st_data, p_st_body_data, p_st_private);
            break;

        case ACTIVEMQ_PRODUCER_INFO:
            p_st_private->is_producer = 1;
            i_ret = openwire_producerinfo_tightUnmarshal(p_st_data, p_st_body_data, p_st_private);
            break;

//        case ACTIVEMQ_TRANSACTION_INFO:
//            i_ret = openwire_transactioninfo_tightUnmarshal(p_st_data, p_st_body_data, p_st_private);
//            break;

//        case ACTIVEMQ_DESTINATION_INFO:
//            break;

//        case ACTIVEMQ_REMOVE_SUBSCRIPTION_INFO:
//            i_ret = openwire_removesubscriptioninfo_tightUnmarshal(p_st_data, p_st_body_data, p_st_private);
//            break;
//
//        case ACTIVEMQ_KEEP_ALIVE_INFO:
//            i_ret = openwire_keepaliveinfo_tightUnmarshal(p_st_data, p_st_body_data, p_st_private);
//            break;
//
        case ACTIVEMQ_SHUTDOWN_INFO:
            p_st_private->i_flag_data_type = FRLOGOUT_END;
            i_ret = openwire_shutdowninfo_tightUnmarshal(p_st_data, p_st_body_data, p_st_private);
            break;
//
//        case ACTIVEMQ_REMOVE_INFO:
//            i_ret = openwire_removeinfo_tightUnmarshal(p_st_data, p_st_body_data, p_st_private);
//            break;
//
//        case ACTIVEMQ_CONTROL_COMMAND:
//            p_st_private->i_flag_detected = 0;
//            i_ret = openwire_controlcommand_tightUnmarshal(p_st_data, p_st_body_data, p_st_private);
//            break;
//
//        case ACTIVEMQ_FLUSH_COMMAND:
//            p_st_private->i_flag_detected = 0;
//            i_ret = openwire_flushcommand_tightUnmarshal(p_st_data, p_st_body_data, p_st_private);
//            break;
//
//        case ACTIVEMQ_CONNECTION_ERROR:
//            i_ret = openwire_connectionerror_tightUnmarshal(p_st_data, p_st_body_data, p_st_private);
//            break;
//
//        case ACTIVEMQ_CONSUMER_CONTROL:
//            i_ret = openwire_consumercontrol_tightUnmarshal(p_st_data, p_st_body_data, p_st_private);
//            break;
//
//        case ACTIVEMQ_CONNECTION_CONTROL:
//            p_st_private->i_flag_detected = 0;
//            i_ret = openwire_connectioncontrol_tightUnmarshal(p_st_data, p_st_body_data, p_st_private);
//            break;

        /*
        * Messages that go back and forth between the client
        * and the server.
        */
//        case ACTIVEMQ_PRODUCER_ACK:
//            i_ret = openwire_producerack_tightUnmarshal(p_st_data, p_st_body_data, p_st_private);
//            break;
//
//        case ACTIVEMQ_MESSAGE_PULL:
//            i_ret = openwire_messagepull_tightUnmarshal(p_st_data, p_st_body_data, p_st_private);
//            break;
//
        case ACTIVEMQ_MESSAGE_DISPATCH:
            p_st_private->is_messages = 1;
            p_st_private->i_flag_report = 1;
            i_ret = openwire_messagedispatch_tightUnmarshal(p_st_data, p_st_body_data, p_st_private);
            break;
//
//        case ACTIVEMQ_MESSAGE_ACK:
//            i_ret = openwire_messageack_tightUnmarshal(p_st_data, p_st_body_data, p_st_private);
//            break;

//        case ACTIVEMQ_ACTIVEMQ_BYTES_MESSAGE:
//            break;

        case ACTIVEMQ_ACTIVEMQ_MAP_MESSAGE:
            p_st_private->is_messages = 1;
            p_st_private->i_flag_report = 1;
            i_ret = openwire_map_message_tightUnmarshal(p_st_data, p_st_body_data, p_st_private);
            break;

        case ACTIVEMQ_ACTIVEMQ_TEXT_MESSAGE:
            p_st_private->is_messages = 1;
            p_st_private->i_flag_report = 1;
            i_ret = openwire_text_message_tightUnmarshal(p_st_data, p_st_body_data, p_st_private);
            break;
    
        case ACTIVEMQ_ACTIVEMQ_STREAM_MESSAGE:
        case ACTIVEMQ_ACTIVEMQ_BYTES_MESSAGE:
        case ACTIVEMQ_ACTIVEMQ_BLOB_MESSAGE:
        case ACTIVEMQ_ACTIVEMQ_OBJECT_MESSAGE:
            p_st_private->is_messages = 1;
            p_st_private->i_flag_report = 1;
            i_ret = openwire_stream_message_tightUnmarshal(p_st_data, p_st_body_data, p_st_private);
            break;
            
        /*
        * Command Response messages
        *
        */
//        case ACTIVEMQ_RESPONSE:
//            p_st_private->i_flag_detected = 0;
//            i_ret = openwire_response_tightUnmarshal(p_st_data, p_st_body_data, p_st_private);
//            break;
//
//        case ACTIVEMQ_EXCEPTION_RESPONSE:
//            i_ret = openwire_exceptionresponse_tightUnmarshal(p_st_data, p_st_body_data, p_st_private);
//            break;

//        case ACTIVEMQ_DATA_RESPONSE:
//        case ACTIVEMQ_DATA_ARRAY_RESPONSE:
//        case ACTIVEMQ_INTEGER_RESPONSE:
//            break;

//        /*
//        * Used by discovery
//        */
//        case ACTIVEMQ_DISCOVERY_EVENT:
//        /*
//        * Command object used by the Journal
//        */
//        case ACTIVEMQ_JOURNAL_ACK:
//        case ACTIVEMQ_JOURNAL_REMOVE:
//        case ACTIVEMQ_JOURNAL_TRACE:
//        case ACTIVEMQ_JOURNAL_TRANSACTION:
//        case ACTIVEMQ_DURABLE_SUBSCRIPTION_INFO:
//            break;
//
//        /*
//        * Reliability and fragmentation
//        */
//        case ACTIVEMQ_PARTIAL_COMMAND:
//        case ACTIVEMQ_PARTIAL_LAST_COMMAND:
//        case ACTIVEMQ_REPLAY:
//            break;
//
//        /*
//        * Broker to Broker command objects
//        */
//        case ACTIVEMQ_MESSAGE_DISPATCH_NOTIFICATION:
//        case ACTIVEMQ_NETWORK_BRIDGE_FILTER:
//            break;

        default:
            p_st_private->is_messages = 0;
            //只解析COMMAND ID responserequired
            return APM_L7_ANALYZE_SUCC;
    }
//    if(p_st_data->ui_remaining != 0)
//        return APM_L7_ANALYZE_FAIL;

    //session 数据上报
//
//if(i_ret == FR_FAIL)
//    printf(" arr_type = %d\n", arr_type);

    return i_ret == FR_SUCC ? APM_L7_ANALYZE_SUCC:APM_L7_ANALYZE_FAIL;

}

/*******************************************************************************
*  FUNC     :  openwire messageack 解析  detect时使用
*  ARGS     :  p_data                       解析数据
*           :  ui_data_len                 数据长度
*  RTN      :  FR_FAIL                     失败
*           :  FR_SUCC                     成功
*  NOTE     :
*******************************************************************************/
int openwire_messageack_detect(char *p_data, unsigned int ui_data_len, Conntrack *p_st_conn)
{
    int i_ret = 0;
    int i_tightEncodingEnabled = 0;
    int i_cacheEnabled = 0;
    Frbyte_Packet *p_st_data = NULL;
    unsigned char uc_tmp = 4;
    int i_openwire_version = 10;

    if(p_data == NULL || ui_data_len == 0 || p_st_conn == NULL)
        return FR_FAIL;

    p_st_data = bytebuff_creat();
    if(p_st_data == NULL)
        return FR_FAIL;

    i_ret = bytebuff_add(p_st_data, p_data, ui_data_len);
    if(i_ret == FR_FAIL)
    {
        bytebuff_free(&p_st_data);
        return FR_FAIL;
    }

    //messageack 解析
    i_ret = FR_FAIL;
    while(i_ret == FR_FAIL && uc_tmp)
    {
        //尝试并清空数据
        uc_tmp--;
        i_tightEncodingEnabled = uc_tmp & '\x01';
        i_cacheEnabled = (uc_tmp >> 1) & '\x01';
        bytebuff_rewind(p_st_data);

        for(i_openwire_version=10;i_openwire_version >=9; i_openwire_version--)
        {
            i_ret = openwire_messageack_tightUnmarshal_detect(p_st_data, i_tightEncodingEnabled, i_cacheEnabled, i_openwire_version);
            if(i_ret == FR_SUCC)
                break;

            bytebuff_rewind(p_st_data);
        }
    }

    bytebuff_free(&p_st_data);
    return i_ret;

}

/*******************************************************************************
*  FUNC     :  openwire messageack 解析  detect时使用
*  ARGS     :  p_data                       解析数据
*           :  ui_data_len                 数据长度
*  RTN      :  FR_FAIL                     失败
*           :  FR_SUCC                     成功
*  NOTE     :
*******************************************************************************/
int openwire_messagedispatch_detect(char *p_data, unsigned int ui_data_len, Conntrack *p_st_conn)
{
    int i_ret = 0;
    int i_tightEncodingEnabled = 0;
    int i_cacheEnabled = 0;
    Frbyte_Packet *p_st_data = NULL;
    unsigned char uc_tmp = 4;
    int i_openwire_version = 10;

    if(p_data == NULL || ui_data_len == 0 || p_st_conn == NULL)
        return FR_FAIL;

    p_st_data = bytebuff_creat();
    if(p_st_data == NULL)
        return FR_FAIL;

    i_ret = bytebuff_add(p_st_data, p_data, ui_data_len);
    if(i_ret == FR_FAIL)
    {
        bytebuff_free(&p_st_data);
        return FR_FAIL;
    }

    //messageack 解析
    i_ret = FR_FAIL;
    while(i_ret == FR_FAIL && uc_tmp)
    {
        //尝试并清空数据
        uc_tmp--;
        i_tightEncodingEnabled = uc_tmp & '\x01';
        i_cacheEnabled = (uc_tmp >> 1) & '\x01';
        bytebuff_rewind(p_st_data);

        for(i_openwire_version = 10;i_openwire_version >= 9; i_openwire_version--)
        {
            i_ret = openwire_messagedispatch_tightUnmarshal_detect(p_st_data, i_tightEncodingEnabled, i_cacheEnabled, i_openwire_version);
            if(i_ret == FR_SUCC)
                break;

            bytebuff_rewind(p_st_data);
        }
    }

    bytebuff_free(&p_st_data);
    return i_ret;

}

